package com.xw.echo.core.pool;

import com.xw.echo.core.ContainerMap;
import com.xw.echo.core.RPCConfigManager;
import com.xw.echo.core.client.ClientConfig;
import com.xw.echo.core.codec.RPCMessageCodec;
import com.xw.echo.core.protocol.Result;
import com.xw.echo.core.register.HostAndPort;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class ConnectPool {
    private final HostAndPort hostAndPort;
    private final Bootstrap bootstrap;
    private final BlockingQueue<Channel> connectionPool;
    private final NioEventLoopGroup worker;
    private final EventLoopGroup initPoolThread;
    private ClientConfig clientConfig = new ClientConfig();
    private final AtomicInteger totalConnections = new AtomicInteger(0);
    private int maxQueueWaitTime;
    private int maxConnections;
    private int maxConnectTimeOut;
    private int workThreadNum;
    private int initConnections;

    public void info() {
        System.out.println("连接池:" + hostAndPort.getHost() + ":" + hostAndPort.getPort() + connectionPool.toString());
    }
    public void initConfig(){
        this.maxQueueWaitTime = clientConfig.getMaxQueueWaitTime();
        this.maxConnections = clientConfig.getMaxConnections();
        this.maxConnectTimeOut = clientConfig.getMaxConnectTimeOut();
        this.workThreadNum = clientConfig.getClientThread();
        this.initConnections = clientConfig.getInitConnections();
    }
    /**
     * 构造函数，用于初始化连接池
     *
     * @param hostAndPort 用于指定要连接的主机和端口
     */
    public ConnectPool(HostAndPort hostAndPort) {
        initConfig();
        // 创建一个阻塞队列作为连接池，大小为最大连接数
        this.connectionPool = new LinkedBlockingQueue<>(maxConnections);
        // 保存主机和端口信息
        this.hostAndPort = hostAndPort;
        // 创建一个NIO事件循环组，线程数量为工作线程数
        this.worker = new NioEventLoopGroup(workThreadNum);
        // 初始化客户端的Bootstrap配置
        this.bootstrap = createrBootstrap();
        this.initPoolThread = new DefaultEventLoopGroup(1);
        initConnectPool();
    }
    private void initConnectPool() {
        initPoolThread.execute(() -> {
            for (int i = 0; i < initConnections; i++) {
                Channel channel = createChannel();
                if (channel != null && channel.isActive()) {
                    connectionPool.add(channel);
                }
            }
        });
    }

    private Bootstrap createrBootstrap() {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(1024, 3, 4, 0, 0));
                        pipeline.addLast(new LoggingHandler());
                        pipeline.addLast(new RPCMessageCodec(RPCConfigManager.getSerializer()));
                        pipeline.addLast(new IdleStateHandler(0,0,180));
                        pipeline.addLast(new HeartbeatHandler());
                        pipeline.addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                if (msg instanceof Result) {
                                    Result result = (Result) msg;
                                    ConcurrentHashMap<String, Promise<Object>> promiseMap = ContainerMap.PromiseMap;
                                    String requestId = result.getRequestId();

                                    Promise<Object> respPromise = promiseMap.get(requestId);
                                    if (respPromise != null) {

                                        if (result.getResultValue()!=null){

                                          respPromise.setSuccess(result.getResultValue());

                                        }
                                        if (result.getException()!=null){
                                            respPromise.setFailure(result.getException());
                                        }
                                    }
                                }

                            }
                        });
                    }
                });

        return bootstrap;
    }
    public Channel getChannel() {
        Channel channel = null;
        while (true) {
            channel = connectionPool.poll();
            if (channel == null) {
                break;
            }
            if (channel.isActive()) {
                return channel;
            } else {
                totalConnections.decrementAndGet();
            }
        }
        // 连接池为空，尝试创建新连接
        if (totalConnections.get() < maxConnections) {
            try {
                channel = createChannel();
                if (channel != null) {
                    totalConnections.incrementAndGet();
                    return channel;
                }
            } catch (Exception e) {
                return null;
            }
        }
        try {
            channel = connectionPool.poll(maxQueueWaitTime, TimeUnit.SECONDS);
            if (channel != null) {
                if (channel.isActive()) {
                    return channel;
                } else {
                    totalConnections.decrementAndGet();
                    return getChannel();
                }
            } else {
                return null;
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("获取连接时线程被中断：{}", e.getMessage());
            return null;
        }
    }

    public void releaseChannel(Channel channel) {
        if (channel != null && channel.isActive()) {
            boolean offered = connectionPool.offer(channel);
            if (offered) {
            } else {
                // 如果队列已满，关闭多余的连接
                channel.close();
                totalConnections.decrementAndGet();
            }
        } else {
            if (channel != null) {
                totalConnections.decrementAndGet();
            }
        }
    }

    public int getTotalConnections() {
        return totalConnections.get();
    }

    public void returnChannel(Channel channel) throws InterruptedException {

        if (channel != null && channel.isActive()) {
            connectionPool.put(channel);
        } else {
            if (channel != null) {
                channel.close();
                totalConnections.decrementAndGet();
            }

        }
    }

    private Channel createChannel() {
        try {
            // 连接到服务器
            ChannelFuture channelFuture = bootstrap.connect(hostAndPort.getHost(), hostAndPort.getPort());

            // 创建一个 Promise 用于超时控制
            Promise<Channel> promise = new DefaultPromise<>(worker.next());
            Channel channel = channelFuture.channel();
            // 添加监听器，当连接完成时，将 Channel 设置到 Promise 中
            channelFuture.addListener(future -> {
//                Channel channel = channelFuture.channel();
                if (future.isSuccess()) {
                    promise.setSuccess(channel); // 连接成功，设置 Channel 到 Promise
                } else {
                    promise.setFailure(future.cause()); // 连接失败，设置失败原因到 Promise
                }
            });

            // 等待连接完成，设置超时时间为 10秒
            if (promise.await(maxConnectTimeOut, TimeUnit.SECONDS)) {
                if (promise.isSuccess()) {
                    return promise.get(); // 连接成功，返回 Channel
                } else {
                    // 连接失败，打印错误信息
                    promise.cause().printStackTrace();
                    return null;
                }
            } else {
                // 超时，取消连接
                channelFuture.cancel(true); // 取消连接尝试
                log.debug("连接超时，已取消连接尝试。");
                return null;
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    public void shutdown() {
        // 关闭连接池中的所有连接
        connectionPool.forEach(channel -> {
            if (channel != null && channel.isActive()) {
                channel.close();
            }
        });

        // 清空连接池
        connectionPool.clear();
        // 关闭工作线程组
        worker.shutdownGracefully();
        // 关闭初始化连接池的线程组
        initPoolThread.shutdownGracefully();

        // 等待所有线程组关闭完成
        worker.terminationFuture().syncUninterruptibly();
        initPoolThread.terminationFuture().syncUninterruptibly();
        // 重置连接计数器
        totalConnections.set(0);
    }


}
